package com.szh.zx.duoxiancheng.chapter1;

import java.util.ArrayDeque;
import java.util.Queue;

/**
 * 线程-生产者和消费者
 */
public class ProducerConsumerModelThread {
    public static void main(String[] args) {
        MyBlockingQueue2<String> queue = new MyBlockingQueue2<>(10);
        new Producer(queue).start();
        new Consumer(queue).start();
    }
}

// 协作队列
// 这里应该使用 Java 提供专门的阻塞队列
 class MyBlockingQueue<E> {
    private Queue<E> queue = null;
    private int limit;

    public MyBlockingQueue(int limit) {
        this.limit = limit;
        queue = new ArrayDeque<>(limit);
    }

    /**
     * 上面说到，synchronized 保护的对象：都有一个锁，一个等待队列，一个条件队列
     * put和take条件不一样，但是都是加入相同的条件队列，由于条件不同但又使用相同的等待队列
     * 所以只能使用notifyAll
     * Java wait、notify机制的局限性：只能有一个条件队列。这使得对于等待条件的分析变得复杂。
     */
    public synchronized void put(E e) throws InterruptedException {
        while (queue.size() == limit) {
            wait();
        }
        queue.add(e);
        notifyAll();
    }

    public synchronized E take() throws InterruptedException {
        while (queue.isEmpty()) {
            wait();
        }
        E e = queue.poll();
        notifyAll();
        return e;
    }


}

 class Producer extends Thread {
    MyBlockingQueue2<String> queue;

    public Producer(MyBlockingQueue2<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        int num = 0;
        try {
            while (true) {
                String task = String.valueOf(num);
                queue.put(task);
                System.out.println("Produce task " + task);
                num++;
                Thread.sleep((int) (Math.random() * 100));

            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

 class Consumer extends Thread {
    MyBlockingQueue2<String> queue;

    public Consumer(MyBlockingQueue2<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        int num = 0;
        try {
            while (true) {
                String take = queue.take();
                System.out.println("handle task" + take);
                Thread.sleep((int) (Math.random() * 100));

            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}